﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Threading;
using UtilZ.Dotnet.Ex.Base;
using UtilZ.Dotnet.Ex.Cache;
using UtilZ.Dotnet.Ex.Communication.Base;
using UtilZ.Dotnet.Ex.Communication.Net.Base;
using UtilZ.Dotnet.Ex.Communication.Net.DataTransfer.Base;
using UtilZ.Dotnet.Ex.DataStruct;
using UtilZ.Dotnet.Ex.Model;
using UtilZ.Dotnet.Ex.ZLog;

namespace UtilZ.Dotnet.Ex.Communication.Net.DataTransfer.Udp
{
    /// <summary>
    /// Udp数据传输类
    /// </summary>
    public partial class UdpDataTransfer : DataTransferAbs
    {
        private readonly static ObjectCache _cache;

        static UdpDataTransfer()
        {
            _cache = new ObjectCache();
        }

        private readonly int _insId;

        private const int CACHE_TIMEOUT_OFFSET_MS = 100;
        private readonly int _sendBufferSize;
        private readonly NetUdpClient _udpClient;
        private readonly NetUDPUnpacker _udpUnpacker;

        private readonly SortedList<int, ConcurrentQueue<TransferDataItem>> _reciveDataPriorityQueueList;
        private readonly bool _enableBlockingCollectionThread;
        private readonly SingleTypeObjectPool<UdpDataTransferInnerRequest> _requestPool = null;
        private readonly BlockingCollectionThread<UdpDataTransferInnerRequest, object> _udpDataTransferRequestDataThreads = null;
        private readonly ThreadEx _receiveDataThread;
        private readonly EventWaitHandleEx _sendDataEventHandler = new EventWaitHandleEx(false, EventResetMode.AutoReset);

        private readonly ConcurrentDictionary<int, TransferDataItem> _sendDataDic = new ConcurrentDictionary<int, TransferDataItem>();
        protected readonly UdpDataTransferDataOutputArgs _receiveDataOutputArgs = new UdpDataTransferDataOutputArgs();

        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="initPara">初始化参数</param>
        public UdpDataTransfer(UdpDataTransferInitPara initPara)
            : base()
        {
            if (initPara == null)
            {
                throw new ArgumentNullException(nameof(initPara));
            }

            this._insId = this.GetHashCode();

            //初始化5个等级容量,常规应该都够了
            this._reciveDataPriorityQueueList = new SortedList<int, ConcurrentQueue<TransferDataItem>>(5);

            this._udpUnpacker = new NetUDPUnpacker(UnpackerUseMode.Monopolize);
            this._udpUnpacker.PackOutput += this.UdpUnpacker_PackOutput;
            var udpInitPara = new NetUdpClientInitPara(this._udpUnpacker, initPara.LocalEndPoint)
            {
                ReceiveDataMode = initPara.ReceiveDataMode,
                ReceiveThreadName = initPara.ReceiveThreadName,
                EnableBroadcast = true,
                SendMillisecondsTimeout = initPara.SendMillisecondsTimeout,
                ReceiveMillisecondsTimeout = initPara.ReceiveMillisecondsTimeout,
                ReceiveBufferSize = initPara.ReceiveBufferSize,
                SendBufferSize = initPara.SendBufferSize,
                SendInterval100NS = initPara.SendInterval100NS,
            };
            this._udpClient = new NetUdpClient(udpInitPara);
            this._sendBufferSize = initPara.SendBufferSize;

            int threadCount = initPara.ThreadCount;
            if (threadCount <= 0)
            {
                threadCount = 1;
            }

            if (threadCount > 1)
            {
                this._enableBlockingCollectionThread = true;
                this._requestPool = new SingleTypeObjectPool<UdpDataTransferInnerRequest>();
                this._udpDataTransferRequestDataThreads = new BlockingCollectionThread<UdpDataTransferInnerRequest, object>(threadCount, $"{NetHelper.GetIPPortString(initPara.LocalEndPoint)}数据传输请求处理线程", this.UdpDataTransferRequestProcessThreadMethod);
            }
            else
            {
                this._enableBlockingCollectionThread = false;
            }

            this._receiveDataThread = new ThreadEx(this.ReceiveDataThreadMethod, $"{NetHelper.GetIPPortString(initPara.LocalEndPoint)}数据发送线程", true);
        }

        private void UdpUnpacker_PackOutput(object sender, DataOutputArgs e)
        {
            if (this._enableBlockingCollectionThread)
            {
                UdpDataTransferInnerRequest request;
                if (!this._requestPool.TryGet(out request))
                {
                    request = new UdpDataTransferInnerRequest();
                }

                request.Copy(e);
                this._udpDataTransferRequestDataThreads.Add(request);
            }
            else
            {
                this.ProcessReceiveData(e.Data, ((NetLinkInfo)e.LinkInfo).RemoteEndPoint);
            }
        }

        private void UdpDataTransferRequestProcessThreadMethod(object tag, UdpDataTransferInnerRequest request)
        {
            this.ProcessReceiveData(request.Data, request.RemoteEndPoint);
            this._requestPool.Add(request);
        }

        private void ProcessReceiveData(BufferData bufferData, EndPoint fromEndPoint)
        {
            try
            {
                if (bufferData.Length < UdpDataTransferMessageAbs.HEAD_SIZE)
                {
                    ZLoger.Warn($"收到的数据长度小于包头长度,忽略数据.");
                    return;
                }

                using (var ms = new MemoryStream(bufferData.Bytes, bufferData.Offset, bufferData.Length))
                {
                    var reader = new BinaryReader(ms);
                    uint sync = reader.ReadUInt32();//同步字
                    if (sync != UdpDataTransferMessageAbs.SYNC)
                    {
                        ZLoger.Warn($"无效的同步字\"0x{ConvertEx.ToHexadecimaString(sync)}\"与期望同步字\"0x{ConvertEx.ToHexadecimaString(UdpDataTransferMessageAbs.SYNC)}\"不匹配,忽略数据.");
                        return;
                    }
                    byte ver = reader.ReadByte();//版本号
                    byte messageType = reader.ReadByte();//消息类型
                    int rid = reader.ReadInt32();//数据长度
                    int dataLength = reader.ReadInt32();//资源标识

                    switch (messageType)
                    {
                        case UdpDataTransferMessageTypes.SEND:
                            this.ProcessSendMessage(bufferData, reader, rid, dataLength, fromEndPoint);
                            break;
                        case UdpDataTransferMessageTypes.ACK:
                            this.ProcessAckMessage(bufferData, reader, rid, dataLength, fromEndPoint);
                            break;
                        case UdpDataTransferMessageTypes.CANCELL:
                            break;
                        case UdpDataTransferMessageTypes.REQ:
                            this.ProcessReqMessage(bufferData, reader, rid, dataLength, fromEndPoint);
                            break;
                        case UdpDataTransferMessageTypes.RES:
                            this.ProcessResMessage(bufferData, reader, rid, dataLength, fromEndPoint);
                            break;
                        default:
                            throw new NotSupportedException($"不支持的数据类型\"{messageType}\"");
                    }
                }
            }
            catch (Exception ex)
            {
                ZLoger.Error(ex);
            }
        }


        private void ProcessResMessage(BufferData bufferData, BinaryReader reader, int rid, int dataLength, EndPoint fromEndPoint)
        {

        }

        private void ProcessReqMessage(BufferData bufferData, BinaryReader reader, int rid, int dataLength, EndPoint fromEndPoint)
        {
            TransferDataItem transferDataItem;
            if (!this._sendDataDic.TryGetValue(rid, out transferDataItem))
            {
                ZLoger.Warn($"收到来自\"192.168.10.11\"请求的数据传输资源SID:{rid}不存在,忽略.");
                return;
            }

            //todo..
        }

        private void ProcessAckMessage(BufferData bufferData, BinaryReader reader, int rid, int dataLength, EndPoint fromEndPoint)
        {

        }
        private void ProcessSendMessage(BufferData bufferData, BinaryReader reader, int rid, int dataLength, EndPoint fromEndPoint)
        {
            var message = new UdpSendMessage() { RID = rid };
            message.Parse(bufferData, reader);

            if (message.SourceType == DataTransferRIDTypeConstant.MEMORY)
            {
                if (message.AttatchData != null && message.AttatchData.Length > 0)
                {
                    //附加数据已把数据传输完成,先回复传输完成消息
                    this.SendAckMessage(rid, UdpAckMessage.COMPLETED, fromEndPoint);

                    //输出数据
                    int noDataLength = UdpDataTransferMessageAbs.HEAD_SIZE + 24;
                    bufferData.Update(bufferData.Offset + noDataLength, bufferData.Length - noDataLength);
                    this._receiveDataOutputArgs.Update(bufferData, fromEndPoint);
                    base.OnRaiseReceiveData(this._receiveDataOutputArgs);
                }
                else
                {
                    //无附加数据,要传输的数据为内存数据,需要进行拉取,先回复确认消息
                    this.SendAckMessage(rid, UdpAckMessage.RECEIVED, fromEndPoint);
                    //todo..
                }
            }
            else
            {
                if (string.IsNullOrWhiteSpace(message.SourceFileName))
                {
                    //理论上不应该进入此分支
                }
                else
                {
                    //要传输的数据为文件,需要进行拉取,先回复确认消息
                    this.SendAckMessage(rid, UdpAckMessage.RECEIVED, fromEndPoint);
                    //todo..
                }
            }





        }

        private void ProcessSendCompletedMessage(BufferData bufferData, BinaryReader reader, int rid, int dataLength, EndPoint fromEndPoint)
        {
            this.RevemoPullTransferDataItem(rid);

            string key = this.GenerateCacheKey(rid);
            var transferDataItem = _cache.Get(key) as TransferDataItem;
            if (transferDataItem != null)
            {
                _cache.Remove(key);
                if (transferDataItem.SendType == ThreadMode.Sync)
                {
                    transferDataItem.SendCompleted();
                }
                else
                {
                    //base.AddTransferDataItem(transferDataItem);
                    base.OnRaiseAsyncSendCompleted(transferDataItem.SID, DataTransferResult.Succes, null);
                }
            }
            else
            {
                ZLoger.Warn($"收到数据传输资源SID:{rid},其对应的传输数据项不存在,忽略.");
            }
        }




        private void SendAckMessage(int rid, byte result, EndPoint remoteEndPoint)
        {
            var ackMessage = new UdpAckMessage() { RID = rid, Result = result };
            BufferData ackBufferData = ackMessage.ToBufferData();
            this._udpClient.SendTo(ackBufferData.Bytes, ackBufferData.Offset, ackBufferData.Length, System.Net.Sockets.SocketFlags.None, remoteEndPoint);
        }

        private void ReceiveDataThreadMethod(ThreadPara threadPara)
        {
            try
            {
                //ConcurrentQueue<TransferDataItem> queue;
                //TransferDataItem transferDataItem;
                //while (!threadPara.Token.IsCancellationRequested)
                //{
                //    lock (this._sendPriorityLevelQueueList)
                //    {
                //        queue = this._sendPriorityLevelQueueList.Values.FirstOrDefault(t => { return t.Count > 0; });
                //    }

                //    if (queue == null)
                //    {
                //        this._sendDataEventHandler.WaitOne(1000);
                //        continue;
                //    }

                //    if (!queue.TryDequeue(out transferDataItem))
                //    {
                //        continue;
                //    }

                //    var udpSendMessage = new UdpSendMessage();
                //    udpSendMessage.MillisecondsTimeout = transferDataItem.SendOptions.MillisecondsTimeout;

                //    switch (transferDataItem.TransferDataType)
                //    {
                //        case TransferDataType.DataSpan:
                //            udpSendMessage.SourceType = UdpDataTransferSIDTypes.MEMORY;
                //            if (transferDataItem.TransferDataBufferData.Length <= this._initPara.SendBufferSize)
                //            {
                //                udpSendMessage.AttatchData =
                //            }
                //            else
                //            {

                //            }
                //            break;
                //        case TransferDataType.BufferData:
                //            if (transferDataItem.BufferData.DataType == BufferDataType.File)
                //            {
                //                udpSendMessage.SourceType = UdpDataTransferSIDTypes.FILE;
                //            }
                //            else
                //            {
                //                udpSendMessage.SourceType = UdpDataTransferSIDTypes.MEMORY;
                //            }
                //            break;
                //        case TransferDataType.TransferData:
                //            udpSendMessage.SourceType = UdpDataTransferSIDTypes.MEMORY;
                //            break;
                //        default:
                //            throw new NotImplementedException($"未实现的传输数据类型\"{transferDataItem.TransferDataType.ToString()}\"");
                //    }
                //}
            }
            catch (Exception ex)
            {
                ZLoger.Error(ex);
            }
        }












        private string GenerateCacheKey(int rid)
        {
            return $"UdpDataTransfer_{this._insId}_{rid}";
        }

        private void CacheEntryRemovedCallback(CacheEntryRemovedArguments arguments)
        {
            if (arguments.RemovedReason == CacheEntryRemovedReason.Expired)
            {
                var transferDataItem = (TransferDataItem)arguments.CacheItem.Value;
                transferDataItem.SendTimeout();
            }
        }

        private void AddPullTransferDataItem(TransferDataItem transferDataItem)
        {
            this._sendDataDic.Add(transferDataItem.SID, transferDataItem);
        }

        private TransferDataItem RevemoPullTransferDataItem(int rid)
        {
            TransferDataItem transferDataItem;
            this._sendDataDic.TryRemove(rid, out transferDataItem);
            return transferDataItem;
        }





        /// <summary>
        /// 开始收发数据
        /// </summary>
        protected override void PrimitiveStart()
        {
            this._udpClient.Start();
            this._receiveDataThread.Start();
        }


        /// <summary>
        /// 让步收发数据
        /// </summary>
        protected override void PrimitiveStop()
        {
            this._udpClient.Stop();
            this._receiveDataThread.Stop();
        }




        /// <summary>
        /// 发送数据
        /// </summary>
        /// <param name="transferDataItem">发送的选项</param>
        protected override void PrimitiveSend(TransferDataItem transferDataItem)
        {
            var option = transferDataItem.SendOptions as UdpTransferDataOptions;
            if (option == null)
            {
                throw new ArgumentException($"udp数据传输选项类型必须为\"{nameof(UdpTransferDataOptions)}\",且不能为null,");
            }

            var message = new UdpSendMessage()
            {
                RID = transferDataItem.SID,
                MillisecondsTimeout = option.MillisecondsTimeout,
                SourceType = transferDataItem.GetSIDType(),
                //SourceName
                SourceDataLength = transferDataItem.GetDataLength(),
                PriorityLevel = option.PriorityLevel,
                //Data=
            };

            if (message.SourceType == DataTransferRIDTypeConstant.FILE)
            {
                message.SourceFileName = Path.GetFileName(transferDataItem.BufferData.FilePath);
                message.AttatchData = null;
                this.AddPullTransferDataItem(transferDataItem);
            }
            else
            {
                message.SourceFileName = null;

                if (message.SourceDataLength <= this._sendBufferSize)
                {
                    message.AttatchData = transferDataItem.GetData();
                }
                else
                {
                    message.AttatchData = null;
                    this.AddPullTransferDataItem(transferDataItem);
                }
            }

            BufferData bufferData = message.ToBufferData();
            transferDataItem.Reset();
            transferDataItem.SendType = ThreadMode.Sync;
            string key = this.GenerateCacheKey(transferDataItem.SID);


            if (option.MillisecondsTimeout > 0)
            {
                int timeoutCount = 0;
                _cache.Set(key, transferDataItem, option.MillisecondsTimeout + CACHE_TIMEOUT_OFFSET_MS, this.CacheEntryRemovedCallback);
                do
                {
                    this._udpClient.Send(bufferData, option);
                    if (transferDataItem.WaitCompleted(option.MillisecondsTimeout))
                    {
                        break;
                    }
                    else
                    {
                        timeoutCount++;
                        if (timeoutCount > option.TimeoutLimit)
                        {
                            _cache.Remove(key);
                            throw new TimeoutException("发送超时");
                        }
                        else
                        {
                            _cache.Get(key);
                        }
                    }
                } while (true);
            }
            else
            {
                _cache.Set(key, transferDataItem);
                this._udpClient.Send(bufferData, option);
                transferDataItem.WaitCompleted();
            }

            this.RevemoPullTransferDataItem(transferDataItem.SID);
            if (transferDataItem.Timeout)
            {
                throw new TimeoutException("发送超时");
            }

            if (transferDataItem.IsCancell)
            {
                throw new OperationCanceledException("用户已取消发送");
            }
        }

        /// <summary>
        /// 异步发送数据
        /// </summary>
        /// <param name="transferDataItem">发送的选项</param>
        protected override void PrimitiveSendAsync(TransferDataItem transferDataItem)
        {
            var option = transferDataItem.SendOptions as UdpTransferDataOptions;
            if (option == null)
            {
                throw new ArgumentException($"udp数据传输选项类型必须为\"{nameof(UdpTransferDataOptions)}\",且不能为null,");
            }

            var message = new UdpSendMessage()
            {
                RID = transferDataItem.SID,
                MillisecondsTimeout = option.MillisecondsTimeout,
                SourceType = transferDataItem.GetSIDType(),
                //SourceName
                SourceDataLength = transferDataItem.GetDataLength(),
                PriorityLevel = option.PriorityLevel,
                //Data=
            };

            if (message.SourceType == DataTransferRIDTypeConstant.FILE)
            {
                message.SourceFileName = Path.GetFileName(transferDataItem.BufferData.FilePath);
                this.AddPullTransferDataItem(transferDataItem);
            }
            else
            {
                if (message.SourceDataLength <= this._sendBufferSize)
                {
                    message.AttatchData = transferDataItem.GetData();
                }
                else
                {
                    message.AttatchData = null;
                    this.AddPullTransferDataItem(transferDataItem);
                }
            }

            BufferData bufferData = message.ToBufferData();
            transferDataItem.Reset();
            transferDataItem.SendType = ThreadMode.Sync;
            string key = this.GenerateCacheKey(transferDataItem.SID);

            if (option.MillisecondsTimeout > 0)
            {
                int timeoutCount = 0;
                _cache.Set(key, transferDataItem, option.MillisecondsTimeout + CACHE_TIMEOUT_OFFSET_MS, this.CacheEntryRemovedCallback);
                do
                {
                    this._udpClient.Send(bufferData, option);
                    if (transferDataItem.WaitCompleted(option.MillisecondsTimeout))
                    {
                        break;
                    }
                    else
                    {
                        timeoutCount++;
                        if (timeoutCount > option.TimeoutLimit)
                        {
                            _cache.Remove(key);
                            throw new TimeoutException("发送超时");
                        }
                        else
                        {
                            _cache.Get(key);
                        }
                    }
                } while (true);
            }
            else
            {
                _cache.Set(key, transferDataItem);
                this._udpClient.Send(bufferData, option);
                transferDataItem.WaitCompleted();
            }

            this.RevemoPullTransferDataItem(transferDataItem.SID);
            if (transferDataItem.Timeout)
            {
                throw new TimeoutException("发送超时");
            }

            if (transferDataItem.IsCancell)
            {
                throw new OperationCanceledException("用户已取消发送");
            }
        }

        /// <summary>
        /// 取消异步发送
        /// </summary>
        /// <param name="rid">资源标识</param>
        protected override void PrimitiveCancellSend(int rid)
        {
            TransferDataItem transferDataItem;
            if (!this._sendDataDic.TryGetValue(rid, out transferDataItem))
            {
                ZLoger.Warn($"收到数据传输资源SID:{rid},其对应的传输数据项不存在,忽略.");
                return;
            }

            transferDataItem.IsCancell = true;
            transferDataItem.SendCompleted();

            string key = this.GenerateCacheKey(rid);
            _cache.Remove(key);
            this.RevemoPullTransferDataItem(rid);

            base.OnRaiseAsyncSendCompleted(rid, DataTransferResult.Cancell, null);
        }



        /// <summary>
        /// Dispose
        /// </summary>
        /// <param name="disposing"></param>
        protected override void Dispose(bool disposing)
        {
            if (this._enableBlockingCollectionThread)
            {
                this._requestPool.Clear();
                this._udpDataTransferRequestDataThreads.Dispose();
            }

            this._udpClient.Dispose();
            this._sendDataEventHandler.Set();
            //this._sendDataThread.Dispose();
            this._sendDataEventHandler.Dispose();
        }
    }



    internal class UdpDataTransferInnerRequest
    {
        /// <summary>
        /// 获取远程终结点
        /// </summary>
        public EndPoint RemoteEndPoint { get; private set; }

        /// <summary>
        /// 获取本地终结点
        /// </summary>
        public EndPoint LocalEndPoint { get; private set; }

        private byte[] _buffer = null;
        private BufferData _bufferData = null;
        public BufferData Data
        {
            get { return this._bufferData; }

        }

        public void Copy(DataOutputArgs args)
        {
            var linkInfo = (NetLinkInfo)args.LinkInfo;

            this.RemoteEndPoint = linkInfo.RemoteEndPoint;
            this.LocalEndPoint = linkInfo.LocalEndPoint;

            if (this._buffer == null || this._buffer.Length != args.Data.Length)
            {
                this._buffer = new byte[args.Data.Length];
                this._bufferData = new BufferData(this._buffer);
            }

            Array.Copy(args.Data.Bytes, args.Data.Offset, this._buffer, 0, args.Data.Length);
            this._bufferData.Update(0, args.Data.Length);
        }
    }



    /// <summary>
    /// 数据传输收到的数据输出事件参数
    /// </summary>
    public class UdpDataTransferDataOutputArgs : DataTransferDataOutputArgs
    {
        /// <summary>
        /// 数据输出来源
        /// </summary>
        public EndPoint FromEndPoint { get; internal set; }


        /// <summary>
        /// 构造函数
        /// </summary>
        public UdpDataTransferDataOutputArgs()
            : base()
        {

        }

        /// <summary>
        /// 更新结果
        /// </summary>
        /// <param name="data">接收到的数据</param>
        /// <param name="fromEndPoint">数据输出来源</param>
        internal void Update(BufferData data, EndPoint fromEndPoint)
        {
            base.Data = data;
            this.FromEndPoint = fromEndPoint;
        }
    }
}
